package com.leal.sink

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord, RecordMetadata}

import java.util.concurrent.Future

class KafkaSink[K, V](createProducer: () => KafkaProducer[K, V]) extends
  Serializable {
  // 这样能够避免运行时产生NotSerializableExceptions异常
  private lazy val producer: KafkaProducer[K, V] = createProducer()

  def send(topic: String, key: K, value: V): Future[RecordMetadata] =
  // 写入Kafka
    producer.send(new ProducerRecord[K, V](topic, key, value))
  def send(topic: String, value: V): Future[RecordMetadata] =
    producer.send(new ProducerRecord[K, V](topic, value))
}

object KafkaSink {

  import scala.collection.JavaConversions._

  def apply[K, V](config: Map[String, Object]): KafkaSink[K, V] = {
    val createProducerFunc: () => KafkaProducer[K, V] = () => {
      // 新建KafkaProducer
      val producer = new KafkaProducer[K, V](config)
      sys.addShutdownHook {
        // 确保在Executor的JVM关闭前
        // Kafka Producer将缓存中的所有信息写入Kafka
        producer.close()
      }
      producer
    }
    new KafkaSink(createProducerFunc)
  }
  def apply[K, V](config: java.util.Properties): KafkaSink[K, V] = apply(config.toMap)
}
